package com.jianbing.service.impl;

import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;
import org.springframework.stereotype.Service;

import com.jianbing.service.NoticeDataWarehouseService;
import com.jianbing.util.HttpClient;

import net.sf.json.JSONObject;

@Service
public class NoticeDataWarehouseServiceImpl implements NoticeDataWarehouseService {

	Logger logger = Logger.getLogger(NoticeDataWarehouseServiceImpl.class);
	private Integer size = 10000;
	private PriorityQueue<String> queue = new PriorityQueue<String>(size);
	private Lock lock = new ReentrantLock();
	private Condition notFull = lock.newCondition();
	private Condition notEmpty = lock.newCondition();

	public NoticeDataWarehouseServiceImpl() {
		Consumer consumer = new Consumer();
		consumer.start();
	}

	/*
	 * 消费者
	 */
	class Consumer extends Thread {
		@Override
		public void run() {
			consumer();
		}
	}

	private void consumer() {
		while (true) {
			lock.lock();
			try {
				while (queue.size() == 0) {
					try {
						notEmpty.await();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				String msg = queue.poll();
				notFull.signal();
				JSONObject json = JSONObject.fromObject(msg);
				String url = json.getString("url");
				String params = json.getString("params");
				HttpClient.doPost(url, params);
				logger.info("通知数据仓库成功" + msg);
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				lock.unlock();
			}
		}
	}

	/**
	 * 
	 * 生产数据仓库通知消息
	 * 
	 * @author cwl
	 * @date 2017年5月13日
	 *
	 * @param msg
	 */
	@Override
	public void producer(String msg) {
		try {
			lock.lock();
			while (queue.size() == size) {
				try {
					logger.info("数据仓库通知队列已满，等待空余空间");
					notFull.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			queue.offer(msg);
			notEmpty.signal();
			logger.info("向数据仓库通知队列插入一条消息，队列剩余空间：" + (size - queue.size()));
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}

	}
}
